/**
 * File: ParallelQueuedController.java
 * Created by: mhaimel
 * Created on: May 4, 2010
 * CVS:  $Id: ParallelQueuedController.java 1.0 May 4, 2010 5:05:35 PM mhaimel Exp $
 */
package uk.ac.ebi.curtain.controller;

import java.io.Serializable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import uk.ac.ebi.curtain.management.InterceptorInfo;
import uk.ac.ebi.curtain.processor.AbstractQueueProcessor;
import uk.ac.ebi.curtain.utils.concurrent.BlockingQueueShutdown;

/**
 * @author mhaimel
 *
 */
public abstract class ParallelQueuedController 
							<I extends Serializable, O extends Serializable, P extends AbstractQueueProcessor<I, O>> 
							extends ParallelController<I, O, P> {
	
	private final AtomicReference<BlockingQueueShutdown<I>> inputQueue = new AtomicReference<BlockingQueueShutdown<I>>(new BlockingQueueShutdown<I>());
	private final AtomicReference<LinkedBlockingQueue<O>> outputQueue = new AtomicReference<LinkedBlockingQueue<O>>(new LinkedBlockingQueue<O>());
	private final AtomicInteger cnt = new AtomicInteger(0);
	
	protected P buildJob(I input, InterceptorInfo info) {
		if(cnt.getAndIncrement() < getQueueParallel()){
			P controller = buildController();
			controller.setInputQueue(getInputQueue());
			controller.setOutputQueue(getOutputQueue());
			submit(controller);
		}
		try {
			getInputQueue().put(input);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
		return null;
		
	};
	
	@Override
	protected void checkResults() {
		processQueue();
		super.checkResults();
	}

	private void processQueue() {
		O res = null;
		while((res = getOutputQueue().poll()) != null){
			interceptorHandle(res);
		}
	}
	
	protected BlockingQueue<O> getOutputQueue() {
		return outputQueue.get();
	}
	
	protected abstract P buildController();
	
	@Override
	protected void processResult(java.util.concurrent.Future<O> res, P job) {
		processQueue();
		super.processResult(res, job);
	};
	
	public void shutDown(boolean force, InterceptorInfo info) {
		queueShutdown();
		super.shutDown(force,info);
	}

	protected BlockingQueueShutdown<I> getInputQueue() {
		return inputQueue.get();
	}
	
	
	protected void queueShutdown(){
		getInputQueue().shutdown();
//		inputQueue.set(new BlockingQueueShutdown<I>());
	}
}
